package com.central.resident.batch_job

import java.util

import com.central.controller.ClueController.JobData
import com.central.controller.{InitClueConditionEntity, InitComplexClueConditionEntity}
import com.central.utils.LoadESData
import com.typesafe.config.Config
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.scalactic.Accumulation.withGood
import org.scalactic._
import spark.jobserver.api.{JobEnvironment, SingleProblem, SparkJob, ValidationProblem}

import scala.util.{Failure, Success, Try}

object ResidentJob extends SparkJob{
  override type JobData = (SparkSession,String)
  override type JobOutput = Unit

  override def runJob(sc: SparkContext, runtime: JobEnvironment, data: JobData): JobOutput ={
    ResidentAnalysis.getResidentResult(data._2,data._1)
  }

  override def validate(sc: SparkContext, runtime: JobEnvironment, config: Config): JobData Or Every[ValidationProblem] = {
    val properties = LoadESData.loadProperties();
    val spark = SparkSession
      .builder()
      .master("spark://192.168.5.180:7077")
      .config("es.nodes", properties.getProperty("elasticsearch.host", "192.168.5.180"))
      .config("es.port", properties.getProperty("elasticsearch.port", "9200"))
      .config("es.mapping.date.rich", "false")
      .getOrCreate()
    val parameterValidatin: JobData Or One[ValidationProblem] =
      Try { //将传入的参数转换成InitClueConditionEntity
        /**
          * json格式:{"jobId":"123", "relationship":"and/or/not", "conditions":[[{"deviceNumber":"61","begintime":1552709939000,"endtime":1555388339000,"didAppear":1,"threshold":1}, {"deviceNumber":"65","begintime":1552709939000,"endtime":1555388339000,"didAppear":1,"threshold":1}],[ {"deviceNumber":"61","begintime":1552709939000,"endtime":1555388339000,"didAppear":1,"threshold":1}, {"deviceNumber":"65","begintime":1552709939000,"endtime":1555388339000,"didAppear":1,"threshold":1}]]}
          */
        import scala.collection.JavaConversions._
        val content = config.getConfig("body")
        val jobId = content.getString("jobId")
        val relationship = content.getString("relationship")
        import scala.collection.JavaConversions._
        val finalConditions: List[List[InitClueConditionEntity]] = content.getAnyRefList("conditions").toList.map(list => list.asInstanceOf[util.ArrayList[_]].toList).map(conditionsNode => {
          conditionsNode.map(map => {
            val conditionMap = map.asInstanceOf[util.Map[String, Object]]
            val deviceNumber = String.valueOf(conditionMap.get("deviceNumber"))
            val beginTime = String.valueOf(conditionMap.get("begintime"))
            val endTime = String.valueOf(conditionMap.get("endtime"))
            val didAppear = String.valueOf(conditionMap.get("didAppear"))
            val threshold = String.valueOf(conditionMap.get("threshold"))
            InitClueConditionEntity(deviceNumber, beginTime.toLong, endTime.toLong, didAppear.toInt, threshold.toInt)
          })
        })
        jobId
      } match {
        case Success(context) => {

          Good((spark, context))
        }
        case Failure(ex) => Bad(One(SingleProblem("body is not exists or json can't parse. error message:" + ex.getMessage)))
      }
    withGood(parameterValidatin) { a => a }
  }
}
